/**********************************************************************************************************************
** Name       : ql_mqtt.c
** Decription : mqtt connect, sub, pub, shadow register process
** Author     : jason_teng
** History    : 2022-12-05 Created
**********************************************************************************************************************/
#include "zq_gl.h"
#include "dtu.h"
#include "mqtt.h"



//  -----------------------------------------------------------------------------------------------

//  -----------------------------------------------------------------------------------------------
static ql_sem_t mqtt_semp;
int zq_ConnectStatus = 0;

mqtt_client_t mqtt_cli = {0};
// struct mqtt_connect_client_info_t client_info = {0};

static int mqtt_pub_success = 0;
static int mqtt_sub_success = 0;

static ql_task_t mqtt_event_task;
char* mqtt_sub_topics[10] = { 0 };

void zq_MqttInit(ql_task_t task)
{
    mqtt_event_task = task;
    ql_rtos_semaphore_create(&mqtt_semp, 0);
}

int ZyIsMqttConnected(void)
{
    return zq_ConnectStatus;
}


//  -----------------------------------------------------------------------------------------------
static void mqtt_connect_result_cb(mqtt_client_t *client, void *arg, mqtt_connection_status_e status)
{
    ZQLOG("mqtt_connect_result_cb status: %d", status);
    // if(status == 0)
    //     mqtt_connected = 1;
    zq_ConnectStatus = (status == 0)? 1 : 0;
    ql_rtos_semaphore_release(mqtt_semp);
}
static void mqtt_disconnect_result_cb(mqtt_client_t *client, void *arg, int status)
{
    ZQLOG("mqtt_disconnect_result_cb status: %d", status);
    // if(status == 0)
    //     mqtt_connected = 1;
    zq_ConnectStatus = (status == 0)? 0 : 1;
    ql_rtos_semaphore_release(mqtt_semp);
}
static void mqtt_state_exception_cb(mqtt_client_t *client)
{
	ZQLOG("mqtt_state_exception_cb mqtt session abnormal disconnect");
	// mqtt_connected = 0;
    zq_ConnectStatus = 0;
    //ql_rtos_semaphore_release(mqtt_semp);
}
static void mqtt_requst_result_cb(mqtt_client_t *client, void *arg,int err)
{  
    if (err == 0) 
        mqtt_pub_success = 1;
    else {
        ZQLOG("mqtt_requst_result_cb err: %d", err);
        mqtt_pub_success = 0;
    }
    ql_rtos_semaphore_release(mqtt_semp);
}
static void mqtt_subscribe_result_cb(mqtt_client_t *client, void *arg,int err)
{
    ZQLOG("mqtt_subscribe_result_cb err: %d", err);
    if (err == 0) mqtt_sub_success = 1;
    else          mqtt_sub_success = 0;
    ql_rtos_semaphore_release(mqtt_semp);
}
int mqtt_subscribe_msg(const char *ptopic, int qos)
{
    int rc;
    if ((ptopic == NULL) || (qos > 2) || (qos < 0)) return ZQ_RET_ERR;

    rc = ql_mqtt_sub_unsub(&mqtt_cli, ptopic, qos, mqtt_subscribe_result_cb, NULL, 1);
    if (rc == MQTTCLIENT_WOUNDBLOCK)
    {
        ZQLOG("======wait subscrible result");
        ql_rtos_semaphore_wait(mqtt_semp, QL_WAIT_FOREVER);

        if (mqtt_sub_success == 1) rc = ZQ_RET_OK;
    }
    return rc;
}
int mqtt_publish_msg(const char *ptopic, const char *ppayload, int qos)
{
	int rc;
    if ((ptopic == NULL) || (ppayload == NULL) || (qos > 2) || (qos < 0)) return ZQ_RET_ERR;
 
	mqtt_pub_success = 0;
    if (strlen(ppayload) > 1024) { // too large only print first 1024 bytes 
        char tmp = ppayload[1025];
        ((char*)ppayload)[1025] = 0;
        ZQLOG("ptopic: %s -- Payload: %s", ptopic, ppayload);
        ((char*)ppayload)[1025] = tmp;
    }
    else
        ZQLOG("ptopic: %s -- Payload: %s", ptopic, ppayload);
    rc = ql_mqtt_publish(&mqtt_cli, ptopic, ppayload, (uint16_t)strlen(ppayload), qos, 0, mqtt_requst_result_cb,NULL);
    if (rc == MQTTCLIENT_WOUNDBLOCK)
    {
        ZQLOG("====mqtt_publish_msg wait mqtt_semp ... ");
        ql_rtos_semaphore_wait(mqtt_semp, QL_WAIT_FOREVER);
    }
    // if(ppayload)
    //     free((void *)ppayload);
    ZQLOG("====mqtt_publish_msg wait mqtt_semp OK");
    if (mqtt_pub_success != 1) // 认为断网
        zq_ConnectStatus = 0;
    return (mqtt_pub_success == 1) ? ZQ_RET_OK : ZQ_RET_ERR;
}
static void mqtt_inpub_data_cb(mqtt_client_t *client, void *arg, int pkt_id, const char *topic, const unsigned char *payload, unsigned short payload_len)
{
   	ql_event_t event;
    char *buf = NULL;
    // char* msgId = NULL;
    for (int i = 0; i < sizeof(mqtt_sub_topics)/sizeof(mqtt_sub_topics[0]); i++) {
        if (mqtt_sub_topics[i] == NULL) break;
        int len = strlen(mqtt_sub_topics[i]);
        if (strncmp(topic, mqtt_sub_topics[i], len) == 0) {
            event.id = ZQ_EVENT_MQTT + i;
            buf = malloc(payload_len + 1);  // 有个0结尾
            if (buf == NULL) return;
            memcpy(buf, payload, payload_len);
            buf[payload_len] = 0;
            ZQLOG("Mqtt received topic: [%s]:%s", topic, buf);
            event.param1 = (size_t)buf;
            event.param2 = payload_len;
            ql_rtos_event_send(mqtt_event_task, &event);
            return;
        }
    }

   	ZQLOG("mqtt received unknown topic: %s, discard", topic);
    return;
}

int ZqMqttRegistTopic(int topic_msg_num, char* topic_str)
{
    int ret = ZQ_RET_OK;
    // static int idx = 0;
    mqtt_sub_topics[topic_msg_num - ZQ_EVENT_MQTT] = topic_str;
    /* sub request topoc_str */
    ret = mqtt_subscribe_msg(topic_str, 1);
    ZQLOG("mqtt subtopic: [%s], rc: [%d]", topic_str, ret);
    return ret;
}

// struct mqtt_connect_client_info_t client_info = {0};
int ZqMqttConnect(const char* host, const char* imei, const char* pkey, const char* dev_type)
{
    int ret;
    // make a encrypt sign    
    static char client_id[36] = { 0 }, user_name[20] = { 0 }, password[36] = { 0 };
    int randomtime = 0;
    srand(ZqGetUTCStample());
    randomtime = rand() % 9000 + 1000;
    sprintf(client_id, "DEV:%s_%s_%d", dev_type, imei, randomtime);
    sprintf(user_name, "%s", imei);
    char encrpttext[64] = { 0 };
    sprintf(encrpttext, "%s%s", client_id, pkey);
    Md5_Encrypt_Data(encrpttext, strlen(encrpttext), password);   // 观察函数内部，几乎不会返回failed
    password[32] = 0;     // md5, 128bit, 16进制32字符
    ZQLOG("Akm_MqttSign [%s]:[%s]:[%s]", client_id, user_name, password);
    // try start data call
    ret = zq_StartDataCall();
    if(ret != ZQ_RET_OK) return ret;
    uint16_t sim_cid;
    ql_bind_sim_and_profile(ZQ_SIM_0, ZQ_PROFILE_IDX, &sim_cid);
    struct mqtt_connect_client_info_t client_info;
    ret = ql_mqtt_client_init(&mqtt_cli, sim_cid);
    // set subscribe call back function 
    ql_mqtt_set_inpub_callback(&mqtt_cli, mqtt_inpub_data_cb, NULL);
    client_info.keep_alive = ZY_MQTT_KEEP_ALIVE;
    client_info.pkt_timeout = 5;
    client_info.retry_times = 3;
    client_info.clean_session = 1;
    client_info.will_qos = 0;
    client_info.will_retain = 0;
    client_info.will_topic = NULL;
    client_info.will_msg = NULL;
    client_info.client_id = client_id;
    client_info.client_user = user_name;
    client_info.client_pass = password;
    client_info.ssl_cfg = NULL;
    ret = ql_mqtt_connect(&mqtt_cli, host, mqtt_connect_result_cb, NULL, &client_info, mqtt_state_exception_cb);
    if (ret == MQTTCLIENT_WOUNDBLOCK) {
        ZQLOG("wait mqtt connect result...");
        ql_rtos_semaphore_wait(mqtt_semp, QL_WAIT_FOREVER);
        if (ZyIsMqttConnected()) {
            ZqSetErrCode(ZQ_ERR_CODE_REGISTER, 0);
            return ZQ_RET_OK;
        }
    }
    else if (ret == MQTTCLIENT_SUCCESS) {
        ZqSetErrCode(ZQ_ERR_CODE_REGISTER, 0);
        return ZQ_RET_OK;
    }
    ZQLOG("@err@: ql_mqtt_connect failed: %d", ret);
    ZqSetErrCode(ZQ_ERR_CODE_REGISTER, 1);
    return ZQ_RET_ERR;
}
void ZyMqttDisconnect(void)
{
    ql_mqtt_disconnect(&mqtt_cli, mqtt_disconnect_result_cb, NULL);
}

